package com.mng.sink;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class JdbcSinkTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<SensorReading> sensorDs = env.addSource(new SensorSourceFromMysql());
        sensorDs.print("input");
        sensorDs.addSink(new SensorSinkToMysql());
        env.execute();
    }

}

class SensorSourceFromMysql extends RichSourceFunction<SensorReading> {

    String driver = "com.mysql.cj.jdbc.Driver";
    String url = "jdbc:mysql://localhost:3306/flinksinktest?serverTimezone=GMT";
    String username = "root";
    String password = "1234";
    Connection connection = null;
    PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        String sql = "select * from sensorinput;";
        //获取执行语句
        ps = connection.prepareStatement(sql);
    }
    @Override
    public void run(SourceContext<SensorReading> sourceContext) throws Exception {
        ResultSet resultSet = ps.executeQuery();
        while(resultSet.next()){
            SensorReading sensorReading = new SensorReading(
                    resultSet.getString("id"),
                    resultSet.getLong("timestamp"),
                    resultSet.getDouble("temperaturre")
            );
            sourceContext.collect(sensorReading);
        }
    }

    @Override
    public void cancel() {

    }


    @Override
    public void close() throws Exception {
        super.close();
        if(connection != null){
            connection.close();
        }
        if (ps != null){
            ps.close();
        }

    }

    //获取mysql连接配置
    public Connection getConnection() {
        try {
            //加载驱动
            Class.forName(driver);
            //创建连接
            connection = DriverManager.getConnection(url, username, password);
        } catch (Exception e) {
            System.out.println("********mysql get connection occur exception, msg = " + e.getMessage());
            e.printStackTrace();
        }
        return connection;
    }
}

class SensorSinkToMysql extends RichSinkFunction<SensorReading>{
    String driver = "com.mysql.cj.jdbc.Driver";
    String url = "jdbc:mysql://localhost:3306/flinksinktest?serverTimezone=GMT";
    String username = "root";
    String password = "1234";
    Connection connection = null;
    PreparedStatement ps = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        String sql = "insert into sensoroutput values (?,?,?);";
        ps = connection.prepareStatement(sql);
    }

    @Override
    public void close() throws Exception {
        super.close();
        ps.close();
        connection.close();
    }

    @Override
    public void invoke(SensorReading value, Context context) throws Exception {
        ps.setString(1,value.getId());
        ps.setLong(2,value.getTimestamp());
        ps.setDouble(3,value.getTemperaturre());
        ps.execute();
    }

    //获取mysql连接配置
    public Connection getConnection() {
        try {
            //加载驱动
            Class.forName(driver);
            //创建连接
            connection = DriverManager.getConnection(url, username, password);
        } catch (Exception e) {
            System.out.println("********mysql get connection occur exception, msg = " + e.getMessage());
            e.printStackTrace();
        }
        return connection;
    }
}
